use std::net;
use std::sync::Arc;
use std::time;

use aspect_proto::msg_queue::{MqDirectDeviceSendProto, MqListProto, MqPullProto};
use ecksport_rpc::registry::HandlerRegistry;
use sea_orm::{DatabaseConnection, SqlErr};
use tokio::sync::oneshot;
use tracing::*;

use aspect_db_common::async_mq::AsyncMqDatastore;
use aspect_db_common::device::DeviceDatastore;
use aspect_db_common::device_msg_queue::DeviceMqDatastore;
use aspect_db_common::user::UserDatastore;
use aspect_db_llmq::query::LlmqDatabase;
use aspect_homeserver_rpc::HomeserverRpcServer;
use aspect_proto::identity::{DeviceAnnProto, DeviceDbProto};
use aspect_proto::user::{UserAddDeviceProto, UserRegProto};
use aspect_svc_common::{logging, make_rt};

mod args;
mod config;
mod constants;
mod jobs;
mod rpc_server;
mod storage;
mod svc;

use config::Config;

use crate::svc::device_mq::DeviceMqServer;
use crate::svc::identity::IdentityServer;
use crate::svc::user::UserServer;

fn main() {
    logging::init();
    info!("starting...");

    let args = argh::from_env::<args::Args>();

    let config = Config {
        listen_port: args.net_port,
        rpc_port: args.rpc_port,
        database_url: args.database_url.expect("init: database url unset"),
        sled_path: args.sled_db_path,
    };

    let rt = make_rt("aspect-homeserverd");
    if let Err(e) = rt.block_on(main_task(&config)) {
        error!(err = %e, "STOPPING");
    }

    info!("exiting...")
}

async fn main_task(config: &Config) -> anyhow::Result<()> {
    // Connect to the database.
    debug!(url = %config.database_url, "connecting to database");
    let dbc = sea_orm::Database::connect(&config.database_url)
        .await
        .expect("init: db connect");

    // Apply migrations, if there are any.
    let n_applied = apply_migrations(&dbc)
        .await
        .expect("init: db apply migrations");
    if n_applied > 0 {
        info!(migrations = %n_applied, "connected, applied migrations");
    } else {
        info!("connected, no migrations to apply");
    }

    // Set up rest of data storage layer.
    let sled_db = sled::open(&config.sled_path)?;
    let storage = Arc::new(storage::open_on_primitives(sled_db, dbc)?);

    // Init the DB now that we've constructed everything we need to.
    let db_init_span = info_span!("dbinit");
    storage
        .exec_startup_tasks()
        .instrument(db_init_span)
        .await?;

    // Set up services.
    // TODO replace all this with the new ecksport RPC stuff
    let ident_svc = IdentityServer::new(storage.clone());
    let user_svc = UserServer::new(storage.clone());
    let dmq_svc = DeviceMqServer::new(storage.clone());

    let mut v0reg = HandlerRegistry::new();
    aspect_proto::core::register_identity_endpoints(ident_svc, &mut v0reg);
    aspect_proto::core::register_user_endpoints(user_svc, &mut v0reg);
    aspect_proto::core::register_msgqueue_endpoints(dmq_svc, &mut v0reg);

    let mut net_builder = ecksport_rpc::builder::ServerBuilder::new_with_name("aspect-homeserver");
    net_builder.add_protocol(aspect_proto::protocol::PROTO_V0, v0reg);

    // Start listening.
    let auth = aspect_proto::auth::AspAuthConfig::new_unauth();
    let lis_addr = net::SocketAddr::from((net::Ipv4Addr::new(0, 0, 0, 0), config.listen_port));
    let lis_handle = net_builder.bind_tcp_listener(auth, lis_addr).await?;
    tokio::time::sleep(time::Duration::from_millis(100)).await;

    // Build and start the RPC server.
    let (stop_tx, stop_rx) = oneshot::channel();
    let rpc_impl = rpc_server::RpcImpl::new(storage, stop_tx);
    let rpc_server = jsonrpsee::server::ServerBuilder::new()
        .build(format!("127.0.0.1:{}", config.rpc_port))
        .await?;
    let rpc_han = rpc_server.start(rpc_impl.into_rpc());

    info!("started!");

    // Wait to be told to exit.
    tokio::select! {
        res = tokio::signal::ctrl_c() => {
            res.expect("main: wait for C-c");
        }
        _ = stop_rx => {
            // nothing?
        }
    }

    info!("received stop signal, beginning shutdown...");
    lis_handle.abort();
    let _ = rpc_han.stop();

    // FIXME this too
    //serv_han.stop();

    Ok(())
}

async fn apply_migrations(dbc: &DatabaseConnection) -> Result<usize, sea_orm::DbErr> {
    use aspect_db_sql_migration::{Migrator, MigratorTrait};

    let pending = Migrator::get_pending_migrations(dbc).await?;
    if !pending.is_empty() {
        Migrator::up(dbc, None).await?;
    }

    Ok(pending.len())
}
